package com.fangsheng.kafkastudy.mashibing;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Test;


import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @创建人: 放生
 * @创建时间: 2021/11/6
 * @描述:
 */
public class Lesson01 {

    @Test
    public void producer() throws ExecutionException, InterruptedException {

        String topic = "msb-items"; // 主题
        Properties p = new Properties();
        p.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.10.11:9092,192.168.10.21:9092,192.168.10.31:9092");
        //kafka  持久化数据的MQ  数据-> byte[]，不会对数据进行干预，双方要约定编解码
        //kafka是一个app：：使用零拷贝  sendfile 系统调用实现快速数据消费
        p.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        p.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //	Server完成 producer request 前需要确认的数量。
        //acks=0时，producer不会等待确认，直接添加到socket等待发送；
        //acks=1时，等待leader写到local log就行；默认是acks=1
        //acks=all或acks=-1时，最严苛的 等待isr中所有副本确认,kafka的broker的数据是一致的
        //（注意：确认都是 broker 接收到消息放入内存就直接返回确认，不是需要等待数据写入磁盘后才返回确认，这也是kafka快的原因）
        p.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(p);
        //现在的producer就是一个提供者，面向的其实是broker，虽然在使用的时候我们期望把数据打入topic
        /*
        msb-items
        2partition
        三种商品，每种商品有线性的3个ID
        相同的商品最好去到一个分区里
         */
        while(true){ // 死循环不停的往kafka发送数据
            for (int i = 0; i < 3; i++) {
                for (int j = 0; j <3; j++) {
//                    kafka 默认是根据 key的hash取模来就决定把数据存到哪一个分区去的，所以想要同一类型的数据存到同一个分区去的话
//                    就需要使用相同的key，注意kafka的偏移量是根据分区来维护的，每一个消费者组的消费者消费了分区的数据，都会独立有自己
//                    在当前分区的offset。
                    ProducerRecord<String, String> record = new ProducerRecord<>(topic, "item"+j,"val" + i);
                    Future<RecordMetadata> send = producer
                            .send(record);
//                    注意kafka的数据发送是异步的，当调用send.get()的方法时候会阻塞
                    RecordMetadata rm = send.get(); // 返回元数据对象
                    int partition = rm.partition(); // 返回分区，是哪一个分区
                    long offset = rm.offset(); // 返回的是偏移量
                    System.out.println("key: "+ record.key()+" val: "+record.value()+" partition: "+partition + " offset: "+offset);

                }
            }
        }
    }


    @Test
    public void consumer01(){

         /*
        kafka-consumer-groups.sh --bootstrap-server node02:9092  --list
         */
        //基础配置
        Properties p = new Properties();
        p.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.10.11:9092,192.168.10.21:9092,192.168.10.31:9092");
        p.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //消费的细节 OOXX 没有的话会创建一个消费者组，组名为ooxx
        p.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"OOXX01");
        //KAKFA IS MQ  IS STORAGE
        p.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//第一次启动，没有offset
        /**
         *
         *  earliest : 表示当前消费运行起来之后，会从分区中的历史最早的开始消费 然后开始纪录offset
         *  latest   ： 表示当前消费者起来之后，不会去消费历史数据，而是当前消费者起来之后，会面如果有数据插入就开始消费，然后 开始纪录offset
         *         "What to do when there is no initial offset in Kafka or if the current offset
         *         does not exist any more on the server
         *         (e.g. because that data has been deleted):
         *         <ul>
         *             <li>earliest: automatically reset the offset to the earliest offset
         *             <li>latest: automatically reset the offset to the latest offset</li>
         *             <li>none: throw exception to the consumer if no previous offset is found for the consumer's group</li><li>anything else: throw exception to the consumer.</li>
         *         </ul>";
         */
        p.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");//自动提交时异步提交，丢数据&&重复数据，设为false后后面就需要手动提交
        //一个运行的consumer ，那么自己会维护自己消费进度
        //一旦你自动提交，但是是异步的
        //1，还没到时间，挂了，没提交，重起一个consuemr，参照offset的时候，会重复消费
        //2，一个批次的数据还没写数据库成功，但是这个批次的offset背异步提交了，挂了，重起一个consuemr，参照offset的时候，会丢失消费

//        p.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"15000");//5秒，当前设置配合开启自动提交的设置一起使用，这里是设置开启自动提交之后，设置延迟多久才提交
//        p.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,""); // POLL 拉取数据，弹性，按需，拉取多少？


        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(p);


        //kafka 的consumer会动态负载均衡
        //Arrays.asList("msb-items") 表示当前消费者订阅主题的集合，一个消费者可以订阅多个主题
        //new ConsumerRebalanceListener() 消费者负载均衡监听器，在kafka当中一个分区的数据只能被同一个消费者组中的消费者消费
        //比如当前有两个分区， 如果已经运行起一个消费者去消费一个主题（这时候这两个分区都归他消费），如果在运行同一个消费者组的消费者去消费同一个主题，
        // kafka会自动将两个分区拆分，让两个消费者的一人消费一个分区，如果再起一个同一个组的消费者，这个时候同一个组的消费者的数量大于分区数
        //那么就会有一个消费者消费不了任何数据，就是空转，new ConsumerRebalanceListener()就是监听消费者分区消费的回调
        consumer.subscribe(Arrays.asList("msb-items"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // 当前回调是有同一个消费者组中有多个消费者运行起来消费，后关闭一个消费者导致kafka自动重新分配消费者消费分区的情况的回调。
                // 比如我当前一个主题有2个分区，然后同一个消费者组有两个消费者A,B,kafka会自动将分区一给A消费
                // 分区二给B消费，如果这个时候关掉B,那么kafka就会自动将原本B消费的分区个A消费，从而监听到有消费者删除从而触发的回调
                System.out.println("---onPartitionsRevoked:");
                Iterator<TopicPartition> iter = partitions.iterator();
                while(iter.hasNext()){
                    System.out.println(iter.next().partition());
                }

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                //当前回调是有同一个消费者组有新的消费者运行起来之，后导致kafka自动分配消费者消费分区的情况的回调。
                //比如我当前一个主题有2个分区，一开始同一个消费者组只有一个个消费者A,这时中途又起了一个个A同组的消费者B,
                // 这个时候kafka会自动把分区拆分一个给B消费，这种重新分配就会触发的回调
                System.out.println("---onPartitionsAssigned:");
                Iterator<TopicPartition> iter = partitions.iterator();

                while(iter.hasNext()){
                    System.out.println(iter.next().partition());
                }


            }
        });
        while(true){ // 测试的时候写的死循环以免主线程结束
            /**
             * 常识：如果想多线程处理多分区
             * 每poll一次，用一个语义：一个job启动
             * 一次job用多线程并行处理分区
             * 且，job应该被控制是串行的
             * 以上的知识点，其实如果你学过大数据
             */
            //微批的感觉
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(0));// 0~n 表示多久的时间去拉取一次数据
            if(!records.isEmpty()){
                Iterator<ConsumerRecord<String, String>> iter = records.iterator();
                while(iter.hasNext()){
                    //因为一个consuemr可以消费多个分区，但是一个分区只能给一个组里的一个consuemr消费
                    ConsumerRecord<String, String> record = iter.next();
                    int partition = record.partition();
                    long offset = record.offset();
                    String key = record.key();
                    String value = record.value();
                    System.out.println("key: "+ record.key()+" val: "+ record.value()+ " partition: "+partition + " offset: "+ offset);
                }
            }



        }

    }




    @Test
    public void consumer(){
        /*
        kafka-consumer-groups.sh --bootstrap-server node02:9092  --list
         */
        //基础配置
        Properties p = new Properties();
        p.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.10.11:9092,192.168.10.21:9092,192.168.10.31:9092");
        p.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //消费的细节 OOXX 没有的话会创建一个消费者组，组名为ooxx
        p.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"OOXX");
        //KAKFA IS MQ  IS STORAGE
        p.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//第一次启动，没有offset
        /**
         *         "What to do when there is no initial offset in Kafka or if the current offset
         *         does not exist any more on the server
         *         (e.g. because that data has been deleted):
         *         <ul>
         *             <li>earliest: automatically reset the offset to the earliest offset
         *             <li>latest: automatically reset the offset to the latest offset</li>
         *             <li>none: throw exception to the consumer if no previous offset is found for the consumer's group</li><li>anything else: throw exception to the consumer.</li>
         *         </ul>";
         */
        p.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");//自动提交时异步提交，丢数据&&重复数据
        //一个运行的consumer ，那么自己会维护自己消费进度
        //一旦你自动提交，但是是异步的
        //1，还没到时间，挂了，没提交，重起一个consuemr，参照offset的时候，会重复消费
        //2，一个批次的数据还没写数据库成功，但是这个批次的offset背异步提交了，挂了，重起一个consuemr，参照offset的时候，会丢失消费

//        p.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"15000");//5秒
//        p.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,""); // POLL 拉取数据，弹性，按需，拉取多少？


        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(p);


        //kafka 的consumer会动态负载均衡
        consumer.subscribe(Arrays.asList("msb-items"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("---onPartitionsRevoked:");
                Iterator<TopicPartition> iter = partitions.iterator();
                while(iter.hasNext()){
                    System.out.println(iter.next().partition());
                }

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("---onPartitionsAssigned:");
                Iterator<TopicPartition> iter = partitions.iterator();

                while(iter.hasNext()){
                    System.out.println(iter.next().partition());
                }


            }
        });


        while(true){
            /**
             * 常识：如果想多线程处理多分区
             * 每poll一次，用一个语义：一个job启动
             * 一次job用多线程并行处理分区
             * 且，job应该被控制是串行的
             * 以上的知识点，其实如果你学过大数据
             */
            //微批的感觉 ，注意每次poll数据，如果一个主题有多个分区，会从多个分区poll数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(0));// 0~n

            if(!records.isEmpty()){
                //以下代码的优化很重要
                System.out.println("-----------"+records.count()+"-------------");
                Set<TopicPartition> partitions = records.partitions(); //每次poll的时候是取多个分区的数据
                //且每个分区内的数据是有序的

                /**
                 * 如果手动提交offset
                 * 1，按消息进度同步提交 (消费一条提交一条)
                 * 2，按分区粒度同步提交 （按分区消费，消费一个分区提交一个分区）
                 * 3，按当前poll的批次同步提交 （当前poll 了的，一次提交）
                 *
                 * 思考：如果在多个线程下
                 * 1，以上1，3的方式不用多线程
                 * 2，以上2的方式最容易想到多线程方式处理，有没有问题？
                 */
                for (TopicPartition partition : partitions) {//遍历分区poll下来的集合
                    List<ConsumerRecord<String, String>> pRecords = records.records(partition);//获取每一个分区的数据集合
//                    pRecords.stream().sorted()
                    //在一个微批里，按分区获取poll回来的数据
                    //线性按分区处理，还可以并行按分区处理用多线程的方式
                    Iterator<ConsumerRecord<String, String>> piter = pRecords.iterator();
                    while(piter.hasNext()){ //迭代每一个分区的数据
                        ConsumerRecord<String, String> next = piter.next();
                        int par = next.partition();
                        long offset = next.offset();
                        String key = next.key();
                        String value = next.value();
                        long timestamp = next.timestamp();


                        System.out.println("key: "+ key+" val: "+ value+ " partition: "+par + " offset: "+ offset+"time:: "+ timestamp);

                        TopicPartition sp = new TopicPartition("msb-items", par);
                        OffsetAndMetadata om = new OffsetAndMetadata(offset);
                        HashMap<TopicPartition, OffsetAndMetadata> map = new HashMap<>();
                        map.put(sp,om);

                        consumer.commitSync(map);//这个是最安全的，每条记录级的更新，第一点 按消息进度同步提交 (消费一条提交一条)
                        //单线程，多线程，都可以
                    }
                    long poff = pRecords.get(pRecords.size() - 1).offset();//获取分区内最后一条消息的offset
                    OffsetAndMetadata pom = new OffsetAndMetadata(poff);
                    HashMap<TopicPartition, OffsetAndMetadata> map = new HashMap<>();
                    map.put(partition,pom);
                    consumer.commitSync( map );//这个是第二种，分区粒度提交offset
                    /**
                     * 因为你都分区了
                     * 拿到了分区的数据集
                     * 期望的是先对数据整体加工
                     * 小问题会出现？  你怎么知道最后一条小的offset？！！！！
                     * 感觉一定要有，kafka，很傻，你拿走了多少，我不关心，你告诉我你正确的最后一个小的offset
                     */

                }
                consumer.commitSync();//这个就是按poll的批次提交offset，第3点

            }



        }





    }


    @Test
    public void consumer02(){
        /*
        kafka-consumer-groups.sh --bootstrap-server node02:9092  --list
         */

        //基础配置
        Properties p = new Properties();
        p.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node02:9092,node03:9092,node01:9092");
        p.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //消费的细节
        p.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"OOXX");
        //KAKFA IS MQ  IS STORAGE
        p.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//第一次启动，米有offset
        /**
         *         "What to do when there is no initial offset in Kafka or if the current offset
         *         does not exist any more on the server
         *         (e.g. because that data has been deleted):
         *         <ul>
         *             <li>earliest: automatically reset the offset to the earliest offset
         *             <li>latest: automatically reset the offset to the latest offset</li>
         *             <li>none: throw exception to the consumer if no previous offset is found for the consumer's group</li><li>anything else: throw exception to the consumer.</li>
         *         </ul>";
         */
        p.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");//自动提交时异步提交，丢数据&&重复数据
        //一个运行的consumer ，那么自己会维护自己消费进度
        //一旦你自动提交，但是是异步的
        //1，还没到时间，挂了，没提交，重起一个consuemr，参照offset的时候，会重复消费
        //2，一个批次的数据还没写数据库成功，但是这个批次的offset背异步提交了，挂了，重起一个consuemr，参照offset的时候，会丢失消费

//        p.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"15000");//5秒
//        p.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,""); // POLL 拉取数据，弹性，按需，拉取多少？


        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(p);


        //kafka 的consumer会动态负载均衡
        consumer.subscribe(Arrays.asList("msb-items"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("---onPartitionsRevoked:");
                Iterator<TopicPartition> iter = partitions.iterator();
                while(iter.hasNext()){
                    System.out.println(iter.next().partition());
                }

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("---onPartitionsAssigned:");
                Iterator<TopicPartition> iter = partitions.iterator();

                while(iter.hasNext()){
                    System.out.println(iter.next().partition());
                }


            }
        });


        /**
         * 以下代码是你再未来开发的时候，向通过自定时间点的方式，自定义消费数据位置
         *
         * 其实本质，核心知识是seek方法
         *
         * 举一反三：
         * 1，通过时间换算出offset，再通过seek来自定义偏移
         * 2，如果你自己维护offset持久化~！！！通过seek完成
         *
         */

        Map<TopicPartition, Long> tts =new HashMap<>();
        //通过consumer取回自己分配的分区 as
        Set<TopicPartition> as = consumer.assignment();//先获取到自己这个消费者消费了哪些分区的集合
        while(as.size()==0){ //需要先拉取下才能显示获取到当前消费者的分区情况
            consumer.poll(Duration.ofMillis(100));
            as = consumer.assignment();
        }

        //自己填充一个hashmap，为每个分区设置对应的时间戳
        for (TopicPartition partition : as) {
//            tts.put(partition,System.currentTimeMillis()-1*1000);
            tts.put(partition,1610629127300L);//为当前消费者所消费的每一个分区设置你需要指定从哪一个时间戳开始消费
        }
        //通过consumer的api，取回timeindex的数据
        Map<TopicPartition, OffsetAndTimestamp> offtime = consumer.offsetsForTimes(tts);


        for (TopicPartition partition : as) {
            //通过取回的offset数据，通过consumer的seek方法，修正自己的消费偏移
            OffsetAndTimestamp offsetAndTimestamp = offtime.get(partition);
            long offset = offsetAndTimestamp.offset();  //不是通过time 换 offset，如果是从mysql读取回来，其本质是一样的
            System.out.println(offset);
            consumer.seek(partition,offset);

        }

        try {
            System.in.read(); //演示的时候在这里先阻塞住，按回车建后在继续。实际开发中不需要
        } catch (IOException e) {
            e.printStackTrace();
        }


        while(true){
            /**
             * 常识：如果想多线程处理多分区
             * 每poll一次，用一个语义：一个job启动
             * 一次job用多线程并行处理分区
             * 且，job应该被控制是串行的
             * 以上的知识点，其实如果你学过大数据
             */
            //微批的感觉
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(0));// 0~n

            if(!records.isEmpty()){
                //以下代码的优化很重要
                System.out.println("-----------"+records.count()+"-------------");
                Set<TopicPartition> partitions = records.partitions(); //每次poll的时候是取多个分区的数据
                //且每个分区内的数据是有序的

                /**
                 * 如果手动提交offset
                 * 1，按消息进度同步提交
                 * 2，按分区粒度同步提交
                 * 3，按当前poll的批次同步提交
                 *
                 * 思考：如果在多个线程下
                 * 1，以上1，3的方式不用多线程
                 * 2，以上2的方式最容易想到多线程方式处理，有没有问题？
                 */
                for (TopicPartition partition : partitions) {
                    List<ConsumerRecord<String, String>> pRecords = records.records(partition);
//                    pRecords.stream().sorted()
                    //在一个微批里，按分区获取poll回来的数据
                    //线性按分区处理，还可以并行按分区处理用多线程的方式
                    Iterator<ConsumerRecord<String, String>> piter = pRecords.iterator();
                    while(piter.hasNext()){
                        ConsumerRecord<String, String> next = piter.next();
                        int par = next.partition();
                        long offset = next.offset();
                        String key = next.key();
                        String value = next.value();
                        long timestamp = next.timestamp();


                        System.out.println("key: "+ key+" val: "+ value+ " partition: "+par + " offset: "+ offset+"time:: "+ timestamp);

                        TopicPartition sp = new TopicPartition("msb-items", par);
                        OffsetAndMetadata om = new OffsetAndMetadata(offset);
                        HashMap<TopicPartition, OffsetAndMetadata> map = new HashMap<>();
                        map.put(sp,om);

                        consumer.commitSync(map);//这个是最安全的，每条记录级的更新，第一点
                        //单线程，多线程，都可以
                    }
                    long poff = pRecords.get(pRecords.size() - 1).offset();//获取分区内最后一条消息的offset



                    OffsetAndMetadata pom = new OffsetAndMetadata(poff);
                    HashMap<TopicPartition, OffsetAndMetadata> map = new HashMap<>();
                    map.put(partition,pom);
                    consumer.commitSync( map );//这个是第二种，分区粒度提交offset
                    /**
                     * 因为你都分区了
                     * 拿到了分区的数据集
                     * 期望的是先对数据整体加工
                     * 小问题会出现？  你怎么知道最后一条小的offset？！！！！
                     * 感觉一定要有，kafka，很傻，你拿走了多少，我不关心，你告诉我你正确的最后一个小的offset
                     */

                }
                consumer.commitSync();//这个就是按poll的批次提交offset，第3点



//                Iterator<ConsumerRecord<String, String>> iter = records.iterator();
//                while(iter.hasNext()){
//                    //因为一个consuemr可以消费多个分区，但是一个分区只能给一个组里的一个consuemr消费
//                    ConsumerRecord<String, String> record = iter.next();
//                    int partition = record.partition();
//                    long offset = record.offset();
//                    String key = record.key();
//                    String value = record.value();
//
//                    System.out.println("key: "+ record.key()+" val: "+ record.value()+ " partition: "+partition + " offset: "+ offset);
//                }
            }



        }





    }





}
